package me.yuzho.study.concurrency;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import static net.mindview.util.Print.*;

@SuppressWarnings("serial")
class PipedQueue extends LinkedBlockingQueue<Character> {
    
}


class SenderWithQueue implements Runnable {
    
    private Random rand = new Random(47);
    private PipedQueue pipedQueue;
    private char c = 'A';
    
    public SenderWithQueue(PipedQueue queue) {
        pipedQueue = queue;
    }

    @Override
    public void run() {
        
        try {
            
            while (!Thread.interrupted()) {
               
                // push a character
                pipedQueue.put(new Character(c++));
                TimeUnit.MILLISECONDS.sleep(rand.nextInt(500));
            }
            
        } catch (InterruptedException e) {
            print(e + " Sender sleep interrupted");
        }
        
        print("send off");
    }
}

class ReceiverWithQueue implements Runnable {
    
    private PipedQueue pipedQueue;
    
    public ReceiverWithQueue(PipedQueue queue) {
        pipedQueue = queue;
    }

    @Override
    public void run() {
        
        try {
            
            while (!Thread.interrupted()) {
               
                // take a character
                Character ch = pipedQueue.take();
                printnb("Read: " + ch.charValue() + ", ");
            }
            
        } catch (InterruptedException e) {
            print(e + " Receiver interrupted");
        }
        
        print("receive off");
    }
}

/** 使用Blocking Queue 而不是管道重写PipedIO
 * @author yzhou-citrix
 *
 */
public class PipedIO2 {
    public static void main(String[] args) throws InterruptedException {
        
        PipedQueue queue = new PipedQueue();
        
        SenderWithQueue sender = new SenderWithQueue(queue);
        
        ReceiverWithQueue receiver = new ReceiverWithQueue(queue);
        
        ExecutorService executorService = Executors.newCachedThreadPool();
        
        executorService.execute(sender);
        executorService.execute(receiver);
        
        TimeUnit.SECONDS.sleep(4);
        
        executorService.shutdownNow();

    }
    
    
}
